package com.example.demo.consumer;

import com.innodealing.offsetskip.annotation.KafkaOffsetSkip;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class KafkaDemoConsumer {

    @KafkaListener(topics = "demo-skip", groupId = "demo-skip-group")
    @KafkaOffsetSkip
    public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        
        try {
            // 模拟业务处理逻辑
            if (value.contains("error")) {
                throw new RuntimeException("业务处理异常: " + value);
            }
            
            System.out.println("处理成功: " + value);
            // 手动确认消息处理成功
            ack.acknowledge();
            
        } catch (Exception e) {
            System.err.println("消息处理失败，1秒后将自动重试: " + value + ", 错误: " + e.getMessage());
            // 不调用 ack.acknowledge()，让重试模板进行重试
            throw e;
        }
    }
}
